package com.zetyun.tiger.connector.sr.v1;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.starrocks.connector.flink.table.StarRocksSinkOptions;
import com.zetyun.rt.meta.jsonschema.annotation.JsonSchemaDescription;
import com.zetyun.rt.meta.jsonschema.annotation.JsonSchemaFormat;
import com.zetyun.rt.meta.jsonschema.annotation.JsonSchemaTitle;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;

import java.util.concurrent.TimeUnit;


/**
 * @author ：zhou gh
 * @date ：2022/01/14 16:44
 */
@Getter
@Setter
@ToString
public class RtSrSinkParameter {

    @JsonSchemaTitle("fe jdbc 信息 ")
    @JsonSchemaDescription("Host of the stream load like: `jdbc:mysql://fe_ip1:query_port,fe_ip2:query_port...`.")
    @JsonProperty(defaultValue = "jdbc:mysql://rdx1:9030,rdx2:9030,rdx3:9030")
//    @JsonProperty(defaultValue = "jdbc:mysql://172.20.2.6:9030,172.20.2.7:9030,172.20.2.8:9030")
    private String jdbcUrl;
    @JsonSchemaTitle("fe http 信息")
    @JsonSchemaDescription("Host of the stream load like: `fe_ip1:http_port;fe_ip2:http_port;fe_ip3:http_port`.")
    @JsonProperty(defaultValue = "rdx1:8030;rdx2:8030;rdx3:8030")
//    @JsonProperty(defaultValue = "172.20.2.6:8030;172.20.2.7:8030;172.20.2.8:8030")
    private String loadUrl;

    @JsonSchemaTitle("用户名")
    @JsonProperty(defaultValue = "root")
    private String userName;

    @JsonSchemaTitle("密码")
    private String passWord = "";


    @JsonSchemaTitle("库名")
    @JsonProperty(defaultValue = "tiger")
    private String dbName;

    @JsonSchemaTitle("表名")
    private String tableName;

    @JsonSchemaTitle("刷新间隔(秒)")
    @JsonProperty(defaultValue = "30")
    private Integer sinkBatchFlushInterval;


    @JsonSchemaTitle("刷新间隔(数据条数)")
    @JsonProperty(defaultValue = "300000")
    private String sinkBatchMaxRows;


    @JsonSchemaTitle("sink语义(`at-least-once` or `exactly-once`)")
    @JsonProperty(defaultValue = "at-least-once")
    private String sinkSemantic;

    @JsonSchemaFormat(JsonSchemaFormat.HIDDEN)
    private String format = "json";


    public StarRocksSinkOptions buildStarRocksSinkOptions() {
        return StarRocksSinkOptions.builder()
                .withProperty("jdbc-url", getJdbcUrl())
                .withProperty("load-url", getLoadUrl())
                .withProperty("username", getUserName())
                .withProperty("password", getPassWord() == null ? "" : getPassWord())
                .withProperty("table-name", getTableName())
                .withProperty("database-name", getDbName())
                .withProperty("sink.properties.format", getFormat())
                .withProperty("sink.properties.strip_outer_array", "true")
                .withProperty("sink.semantic", getSinkSemantic())
                .withProperty("sink.buffer-flush.max-rows", getSinkBatchMaxRows())
                .withProperty("sink.buffer-flush.interval-ms", (getSinkBatchFlushInterval() * 1000) + "")
                .build();
    }
}
